/*
============================================================================
 Name        : dd_msg_queue.c
 Author      : Jian
 Version     :
 Copyright   : GPL
 Description : dd_msg_queue.c - source file
============================================================================
*/

/* INCLUDE FILES */
#include "dd_msg_queue.h"
#include "dd_condition.h"
#include "dd_queue.h"
#include "dd_util.h"
#include "sd_platform.h"
#include "dd_log.h"

typedef struct {
    DD_UINT  id ;
    void*    data ;
    DD_UINT  len ;
    sd_semp* waitLock ;
} dd_msg;

static dd_msg* _msg_create(DD_UINT id, void* data, DD_UINT len, sd_semp* waiter) 
{
    dd_msg* msg = (dd_msg*)dd_malloc(sizeof(dd_msg) + len) ;

    __DD_CHECK_MEM(msg, sizeof(dd_msg) + len, return NULL) ;

    msg->id     = id ;
    msg->len    = len ;
    
    if (data && len > 0) {
        msg->data   = (char*)msg + sizeof(dd_msg);
        memcpy(msg->data, data, len) ;
    } else {
        msg->data = data ;
    }

    msg->waitLock = waiter ;

    return msg ;
}

static void _msg_destroy(dd_msg* msg) 
{
    __DD_ASSERT_DEFENSE(msg, return) ;
    dd_free(msg) ;
}


dd_msg_queue* dd_mqueue_create(data_free* freeFun) 
{
    dd_msg_queue* msgQueue = (dd_msg_queue*)dd_malloc(sizeof(dd_msg_queue)) ;
    
    __DD_CHECK_MEM(msgQueue, sizeof(dd_msg_queue), return NULL) ;
    
    msgQueue->freeFun   = freeFun ;
    msgQueue->queueLock = sd_create_sem(1) ;
    msgQueue->queue   = dd_queue_create() ;
    msgQueue->cont    = dd_cont_create(msgQueue->queueLock) ;
    
    __DD_ASSERT_DEFENSE(msgQueue->queueLock && msgQueue->queueLock && msgQueue->cont, return NULL) ;

    return msgQueue ;
}

void dd_mqueue_destroy(dd_msg_queue* msgQueue) 
{
    __DD_ASSERT_DEFENSE(msgQueue, return) ;
    
    /* TODO: more job for thread safe */ 
    
    dd_wait_sem(msgQueue->queueLock, ;) ;
    
    dd_queue_destroy(msgQueue->queue, msgQueue->freeFun) ;
    msgQueue->queue = NULL ;

    dd_cont_destroy(msgQueue->cont) ;
    msgQueue->cont = NULL ;

    dd_post_sem(msgQueue->queueLock, ;) ;

    sd_destroy_sem(msgQueue->queueLock) ;

    dd_free(msgQueue) ;
}

int dd_mqueue_send(dd_msg_queue* msgQueue, DD_UINT id, void* data, DD_UINT len, int flag) 
{
    dd_msg*  msg = NULL ;
    sd_semp* locker = NULL ;
    
    if (flag & EMQFlagWait) 
        locker = sd_create_sem(0) ;
    
    msg = _msg_create(id, data, len, locker) ;

    dd_wait_sem(msgQueue->queueLock, return -1) ;
    
    dd_queue_enqueue(msgQueue->queue, msg) ;

    dd_post_sem(msgQueue->queueLock, return -1) ;
    
    if (flag & EMQFlagSignalAll) 
        dd_cont_signal_all(msgQueue->cont) ;
    else 
        dd_cont_signal(msgQueue->cont) ;

    if (locker) {
        dd_wait_sem(locker, ;) ;
        sd_destroy_sem(locker) ;
    }

    return 0 ;
}

int dd_mqueue_recv(dd_msg_queue* msgQueue, handle_mq_msg* handler, int flag) 
{
    int     ret  = EMQAgain ;
    dd_msg* msg = NULL ;
    
    dd_wait_sem(msgQueue->queueLock, return NULL) ;

    if (!dd_queue_is_empty(msgQueue->queue))
        msg = (dd_msg*)dd_queue_dequeue(msgQueue->queue) ;

    if (!msg && (flag & EMQFlagWait)) {
        do {
            dd_cont_wait(msgQueue->cont) ;
            
            if (!dd_queue_is_empty(msgQueue->queue))
                msg = (dd_msg*)dd_queue_dequeue(msgQueue->queue) ;

        } while (!msg);
    }

    dd_post_sem(msgQueue->queueLock, return NULL) ;

    if (msg) {
        handler(msg->id, msg->data, msg->len) ;
        if (msg->waitLock)
            dd_post_sem(msg->waitLock, ;) ;
        _msg_destroy(msg) ;
        ret = 0 ;
    }

    return ret ;
}

void dd_mqueue_reset(dd_msg_queue* msgQueue) 
{
    dd_wait_sem(msgQueue->queueLock, return) ;
    
    dd_queue_reset(msgQueue->queue, msgQueue->freeFun) ;

    dd_post_sem(msgQueue->queueLock, return) ;
}

/* End of File */
